package com.zcxk.ws;

import com.alibaba.fastjson2.JSON;
import com.zcxk.config.AuthToken;
import com.zcxk.constant.Constants;
import com.zcxk.dto.DeviceData;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.vertx.config.ConfigRetriever;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.http.*;
import io.vertx.core.parsetools.RecordParser;
import io.vertx.ext.web.Router;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class WSApp extends AbstractVerticle {

  private final static Logger logger = LoggerFactory.getLogger(WSApp.class);
  /**
   * All handlers. 表示所有的ws连接
   */
  private final Map<String, Boolean> handlersSet = new ConcurrentHashMap<>(1000);
  /**
   * userId -> set of handlers. 一个用户可能会有多个ws连接，多端
   */
  private final Map<Long, Set<String>> userToHandlersMap = new ConcurrentHashMap<>(1000);
  /**
   * handler -> userId.
   */
  private final Map<String, Long> handlerToUserMap = new ConcurrentHashMap<>(1000);
  /**
   * 主机地址
   */
  private static final String SERVER_HOST = Constants.getServerHost();
  /**
   * 最大重连次数
   */
  private static final int maxReconnectRetries = Constants.getMaxReconnectRetries();
  /**
   * WS服务器端口号
   */
  private final int wsServerPort = Constants.getWsServerPort();
  /**
   * DcpWebServer的 WS服务器端口号
   */
  private static final int dcpWsServerPort = Constants.getDcpWsServerPort();

  public static void main(String[] args) {
    // 使用log4j2
    System.setProperty("vertx.logger-delegate-factory-class-name", "io.vertx.core.logging.Log4j2LogDelegateFactory");
    Vertx vertx = Vertx.vertx();
    WSApp verticle = new WSApp();
    // 部署verticle，这里会调用start方法
    vertx.deployVerticle(verticle);
  }

  @Override
  public void start() {
    ConfigRetriever retriever = ConfigRetriever.create(vertx);
    retriever.getConfig(json -> {
      // 创建VertX server:
      HttpServer server = vertx.createHttpServer(new HttpServerOptions().setRegisterWebSocketWriteHandlers(true));
      // 创建路由:
      Router router = Router.router(vertx);
      // 处理请求 GET /login:
      router.get("/login").handler(ctx -> {
        HttpServerRequest request = ctx.request();
        String tokenStr = request.getParam("token");
        Long userId = null;
        try {
          AuthToken token = AuthToken.fromSecureString(tokenStr, Constants.getSecretKey());
          if (!token.isExpired()) {
            userId = token.userId();
          }
        } catch (IllegalArgumentException exception) {
          ctx.response().setStatusCode(HttpResponseStatus.UNAUTHORIZED.code())
            .setStatusMessage("token invalid")
            .end();
          logger.info("token invalid, token: {}", tokenStr);
          return;
        }
        logger.info("parse user id from token, userId: {}", userId);
        if (Objects.isNull(userId)) {
          ctx.response().setStatusCode(HttpResponseStatus.UNAUTHORIZED.code())
            .setStatusMessage("ws login fail, token is expired")
            .end();
          return;
        }
        Long finalUserId = userId;
        //将HTTP升级为WS
        request.toWebSocket(ar -> {
          if (ar.succeeded()) {
            initWebSocket(ar.result(), finalUserId);
          }
        });
      });
      // 处理请求 GET /actuator/health:健康检查
      router.get("/actuator/health").respond(ctx -> ctx.response().putHeader("Content-Type", "application/json").end("{\"status\":\"UP\"}"));
      // 对于未定义的路由，返回 404 状态码，表示请求的路径不存在。
      router.get().respond(ctx -> ctx.response().setStatusCode(404).setStatusMessage("No Route Found").end());
      // 设置请求处理器并启动服务器，监听指定的端口。
      // WS监听8081
      server.requestHandler(router).listen(this.wsServerPort, result -> {
        if (result.succeeded()) {
          logger.info("Vertx started on port(s): {} (http) with context path ''", this.wsServerPort);
        } else {
          logger.error("Start http server failed on port " + this.wsServerPort, result.cause());
          vertx.close();
          System.exit(1);
        }
      });
    });
    createWebSocketClient(vertx, ConnectionBizType.ECG, this);
  }

  private static void createWebSocketClient(Vertx vertx, ConnectionBizType bizType, WSApp push) {
    final HttpClient httpClient = vertx.createHttpClient();
    final String requestURI = "ws://" + SERVER_HOST + ":" + dcpWsServerPort;

    httpClient.webSocket(dcpWsServerPort, SERVER_HOST, requestURI, conn -> {
      if (conn.succeeded()) {
        final WebSocket socket = conn.result();
        logger.info("WebSocket connected, biz: {}", bizType.name());

        final RecordParser parser = RecordParser.newDelimited("#_#", buffer -> {
          String msg = buffer.toString();
          logger.info("receive data: {}", msg);
          DeviceData deviceData = JSON.parseObject(msg, DeviceData.class);
          deviceData.setFirstTime(System.currentTimeMillis());
          if (deviceData.getNibpReview() != null) {
            logger.info("NIBP：" + deviceData.getDeviceId());
          }
          sendDeviceData(push, deviceData);
        });

        socket.handler(parser);

        // 添加关闭连接后的重试逻辑
        socket.closeHandler(v -> {
          logger.warn("WebSocket closed, attempting to reconnect...");
          attemptReconnect(1, vertx, bizType, push); // 从重试次数1开始重连
        });
      } else {
        // 连接失败的处理，等待后重试
        Throwable throwable = conn.cause();
        logger.error("Failed to connect WebSocket: {}, attempting to reconnect...", throwable.getMessage());
        attemptReconnect(1, vertx, bizType, push); // 从重试次数1开始重连
      }
    });
  }

  /**
   * server断开后自动重连方案
   */
  private static void attemptReconnect(int retry, Vertx vertx, ConnectionBizType bizType, WSApp push) {
    if (retry > maxReconnectRetries) {
      logger.error("reconnect retry {} times, also can not connect to the server", maxReconnectRetries);
    } else {
      long backoff = (long) (Math.pow(2, retry) * 10);

      vertx.setTimer(backoff, timer -> {
        createWebSocketClient(vertx, bizType, push); // 尝试重新连接
      });
    }
  }

  /**
  * 将 DeviceData 对象序列化为 JSON 字符串，并通过 WSApp 实例 push 广播该数据。
  */
  private static void sendDeviceData(WSApp push, DeviceData deviceData) {
    String deviceId = deviceData.getPatientID();
    String prefix = deviceId.substring(0, deviceId.length() - 1); // 获取设备 ID 前缀（如果需要）
    String jsonData = JSON.toJSONString(deviceData); // 将设备数据转换为 JSON 字符串
    push.broadcastDeviceData("biz", jsonData); // 发送设备数据
  }

  /**
   * 将设备数据通过 broadcast 方法发送给所有注册的处理程序。
   */
  public void broadcastDeviceData(String name, String msg) {
    this.broadcast(msg); // 调用通用的广播方法
  }

  /**
   * 向所有注册的处理程序发送文本消息。
   */
  public void broadcast(String text) {
    EventBus eb = vertx.eventBus();
    for (String handler : this.handlersSet.keySet()) {
      eb.send(handler, text); // 向所有处理程序发送消息
    }
  }

  private void initWebSocket(ServerWebSocket websocket, Long userId) {
    String handlerId = websocket.textHandlerID();
    logger.info("websocket accept userId: " + userId + ", handlerId: " + handlerId);
    // handle text message:
    websocket.textMessageHandler(str -> {
      logger.info(str);
      websocket.writeTextMessage("received: " + str);
    });
    websocket.exceptionHandler(t -> {
      logger.error("websocket error: " + t.getMessage(), t);
    });
    // on close:
    websocket.closeHandler(e -> {
      unsubscribeClient(handlerId);
      unsubscribeUser(handlerId, userId);
      logger.info("websocket closed: " + handlerId);
    });
    subscribeClient(handlerId);
    subscribeUser(handlerId, userId);
    // send welcome message:
    if (userId == null) {
      websocket.writeTextMessage(
        "{\"type\":\"status\",\"status\":\"connected\",\"message\":\"connected as anonymous user\"}");
    } else {
      websocket.writeTextMessage(
        "{\"type\":\"status\",\"status\":\"connected\",\"message\":\"connected as user\",\"userId\":"
          + userId + "}");
    }
  }

  private void subscribeClient(String handlerId) {
    this.handlersSet.put(handlerId, Boolean.TRUE);
  }

  private void unsubscribeClient(String handlerId) {
    this.handlersSet.remove(handlerId);
  }

  private void subscribeUser(String handlerId, Long userId) {
    if (userId == null) {
      return;
    }
    handlerToUserMap.put(handlerId, userId);
    Set<String> set = userToHandlersMap.computeIfAbsent(userId, k -> new HashSet<>());
    set.add(handlerId);
    logger.info("subscribe user {} {} ok.", userId, handlerId);
  }

  private void unsubscribeUser(String handlerId, Long userId) {
    if (userId == null) {
      return;
    }
    handlerToUserMap.remove(handlerId);
    Set<String> set = userToHandlersMap.get(userId);
    if (set != null) {
      set.remove(handlerId);
      if (set.isEmpty()) {
        userToHandlersMap.remove(userId);
        logger.info("unsubscribe user {} {} ok: cleared.", userId, handlerId);
      } else {
        logger.info("unsubscribe user {} {} ok: but still others online.", userId, handlerId);
      }
    }
  }
}
